package com.zysd.collection.scheduler.system.config;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.joda.JodaModule;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;

/**
 * 售电采集RabbitMQ 配置
 *
 * @author liuht
 */
@Configuration
@EnableConfigurationProperties(MasterConfig.class)
public class AmqpConfig {
    /**
     * 日志
     */
    private Logger logger = LoggerFactory.getLogger(AmqpConfig.class);

    @Value("${spring.rabbitmq.host:}")
    private String host;
    @Value("${spring.rabbitmq.addresses:}")
    private String addresses;
    @Value("${spring.rabbitmq.username:guest}")
    private String username;
    @Value("${spring.rabbitmq.password:guest}")
    private String password;
    @Value("${spring.rabbitmq.channelcachesize:25}")
    private String channelCacheSize;
    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;


    @Value("${collectd.meters.rabbitmq.send-exchange}")
    private String sendExchange;

    @Value("${collectd.meters.rabbitmq.send-routingKey}")
    private String sendRoutingKey;

    @Value("${collectd.meters.rabbitmq.send-queue}")
    private String sendQueue;

    @Autowired
    private MasterConfig masterConfig;


    /**
     * RabbitMq链接工厂
     *
     * @return ConnectionFactory
     */
    @Bean
    @Primary
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        /*Proxy类主机信息*/
        if (!StringUtils.isEmpty(host)) {
            connectionFactory.setHost(host);
        }
        /*多点接入主机列表*/
        if (!StringUtils.isEmpty(addresses)) {
            connectionFactory.setAddresses(addresses);
        }
        /*连接数量*/
        connectionFactory.setChannelCacheSize(Integer.valueOf(channelCacheSize));
        /**/
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.addConnectionListener(new ConnectionListener() {
            @Override
            public void onCreate(Connection connection) {
                logger.info("创建rabbitmq连接");
            }

            @Override
            public void onClose(Connection connection) {
                logger.info("关闭rabbitmq连接");
            }
        });
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }


    /**
     * 配置MQ传输序列化对象
     *
     * @return Jackson2JsonMessageConverter
     */
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        ObjectMapper mapper = new ObjectMapper();
        mapper.registerModule(new JodaModule());
        mapper.enable(MapperFeature.USE_GETTERS_AS_SETTERS);
        mapper.enable(MapperFeature.ALLOW_FINAL_FIELDS_AS_MUTATORS);
        mapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
        mapper.enable(SerializationFeature.WRITE_ENUMS_USING_TO_STRING);
        mapper.enable(DeserializationFeature.FAIL_ON_NUMBERS_FOR_ENUMS);
        mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
        mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        return new Jackson2JsonMessageConverter(mapper);
    }

    /**
     * 配置模版对象
     *
     * @return RabbitTemplate
     */
    @Bean
    public RabbitTemplate template() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }

    /**
     * 配置管理器
     *
     * @return RabbitAdmin
     */
    @Bean
    public RabbitAdmin rabbitAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setIgnoreDeclarationExceptions(true);
        return rabbitAdmin;
    }

    @Bean("sendExchange")
    public DirectExchange sendExchange() {
        return new DirectExchange(sendExchange);
    }

    @Bean
    public Object declareQueue() {
        masterConfig.getMasterServers().forEach(masterServer -> {
            String ipPort = masterServer.replaceAll(":", ".");
            String send = sendQueue + "." + ipPort;
            String sendRoutingKeys = sendRoutingKey + "." + ipPort;
            declare(send, sendRoutingKeys, sendExchange());
        });
        return null;
    }

    private void declare(String queueName, String routingKey, DirectExchange exchange) {
        RabbitAdmin admin = rabbitAdmin();
        Queue queue = new Queue(queueName, true, false, false);
        admin.declareQueue(queue);
        admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey));
    }

    /**
     * RepublishMessageRecoverer允许在重试次数耗尽后，发布失败消息
     *
     * @return RetryOperationsInterceptor
     */
    @Bean
    public RetryOperationsInterceptor interceptor() {
        return RetryInterceptorBuilder.stateless()
                .maxAttempts(5)
                .recoverer(new RepublishMessageRecoverer(template()))
                .build();
    }
}

